
Err rp_syslog();


Err rp_cqueue_init(Cqueue *cqueue, int qlength);
Err rp_cqueue_finalize(Cqueue *cqueue);
Err rp_cqueue_put(Cqueue *cqueue, Cmd *cmd);
Cmd* rp_cqueue_get(Cqueue *cqueue);

Err rp_tqueue_init(Tqueue *tqueue, int qlength);
Err rp_tqueue_finalize(Tqueue *tqueue);
Err rp_tqueue_put(Tqueue *tqueue, Task *task);
Task* rp_tqueue_get(Tqueue *tqueue);


/*=============		Implementation 		==============*/
Err rp_cqueue_init(Cqueue *cqueue, int qlength)
{
	char fname[] = "rc_cqueue_init";
	char msg[] = "fatal error";
	Err err;

	err = rp_mutex_lock(cqueue->mutex);
	
	cqueue->qlength = qlength;
	cqueue->ncmd = 0;	
	cqueue->head = NULL;
	cqueue->tail = NULL;
	
	err = rp_mutex_unlock(cqueue->mutex);

	return err;
}

Err rp_cqueue_finalize(Cqueue *cqueue)
{
	char fname[] = "rc_cqueue_destroy";
	char msg[] = "fatal error";
	Err err;	
	Cnode *cnode;
	
	err = rp_mutex_lock(&(cqueue->mutex));
	if (!cqueue->qlength)	
	{
		rp_log(LOG_DEBUG, fname, "The command queue has NOT been initialized");
		err = rp_mutex_unlock(&(cqueue->mutex));
		return err;			
	}
	if (!cqueue->ncmd)	
	{
		rp_log(LOG_DEBUG, fname, "The command queue is empty");
		err = rp_mutex_unlock(&(cqueue->mutex));
		return err;			
	}
	
	cnode = cqueue->tail;
	while (cnode->prev)
	{
		cnode=cnode->prev;
		free(cnode->next);
	}
	free(cnode);

	cqueue->head = NULL;
	cqueue->tail = NULL;
	cqueue->qlength = 0;
	cqueue->ncmd = 0;
	
	err = rp_mutex_unlock(&(cqueue->mutex));
	
	return err;
}


Err rp_cqueue_put(Cqueue *cqueue, Cmd *cmd)
{
	char fname[] = "rc_cqueue_put";
	char msg[] = "fatal error";
	Err err;
	Cnode *cnode;

	err = rp_mutex_lock(&(cqueue->mutex));
	if (!cqueue->qlength)	
	{
		rp_log(LOG_DEBUG, fname, "The command queue has NOT been initialized");
		err = rp_mutex_unlock(&(cqueue->mutex));
		return err;			
	}
	if (cqueue->ncmd ==  cqueue->qlength)	
	{
		rp_log(LOG_INFO, fname, "The command queue is full");	
		err = rp_mutex_unlock(&(cqueue->mutex));
		return err;			
	}
	
	if (cqueue->ncmd == 0)
	{
		cnode = (Cnode*)malloc(sizeof(Cnode));
		cnode->next = NULL;
		cnode->prev = NULL;
		memcpy(&(cnode->cmd), cmd, sizeof(Cmd));
		
		cqueue->head = cnode;
		cqueue->tail = cnode;
		cqueue->ncmd++;
	} else {		// 1<=ncmd<qlength
		cnode = (Cnode*)malloc(sizeof(Cnode));
		cnode->next = NULL;
		cnode->prev = cqueue->tail;
		memcpy(&(cnode->cmd), cmd, sizeof(Cmd));
				
		cqueue->tail->next = cnode;
		cqueue->tail = cnode;
		cqueue->ncmd++;
		
	}
	
	err = rp_mutex_unlock(&(cqueue->mutex));

	return err;
}


Cmd* rp_cqueue_get(Cqueue *cqueue)
{
	char fname[] = "rc_cqueue_get";
	char msg[] = "fatal error";
	Err err;	
	Cnode *cnode;
	Cmd *cmd;
	
	err = rp_mutex_lock(&(cqueue->mutex));
	if (cqueue->ncmd == 0 || cqueue->qlength ==0) return err; 		// TODO: the queue is empty or the queue has not been initialized
	if (!cqueue->qlength)	
	{
		rp_log(LOG_DEBUG, fname, "The command queue has NOT been initialized");
		err = rp_mutex_unlock(&(cqueue->mutex));
		return err;			
	}
	if (!cqueue->ncmd)	
	{
		rp_log(LOG_DEBUG, fname, "The command queue is empty");	
		err = rp_mutex_unlock(&(cqueue->mutex));
		return err;			
	}
	
	cnode = cqueue->head;
	cmd = (Cmd*)malloc(sizeof(Cmd));
	memcpy(cmd, cnode->cmd);
	
	if (cqueue->tail == cqueue->head) 
	{
		cqueue->head = NULL;
		cqueue->tail = NULL;
	} else {
		cqueue->head = cqueue->head->next;
		cqueue->head->prev = NULL;
	}
	free(cnode);
	
	err = rp_mutex_unlock(&(cqueue->mutex));

	return cmd;
}

/*  ============== 		Task Queue		=================*/

Err rp_tqueue_init(Tqueue *tqueue, int qlength)
{
	char fname[] = "rc_tqueue_init";
	char msg[] = "fatal error";
	Err err;

	err = rp_mutex_lock(tqueue->mutex);
	
	tqueue->qlength = qlength;
	tqueue->ntask = 0;	
	tqueue->head = NULL;
	tqueue->tail = NULL;
	
	err = rp_mutex_unlock(tqueue->mutex);

	return err;
}

Err rp_tqueue_finalize(Tqueue *tqueue)
{
	char fname[] = "rc_tqueue_destroy";
	char msg[] = "fatal error";
	Err err;	
	Tnode *tnode;

	if (!tqueue->qlength)	return err;			// The queue has NOT been initialized

	err = rp_mutex_lock(&(tqueue->mutex));
	if (!tqueue->qlength)	
	{
		rp_log(LOG_DEBUG, fname, "The task queue has NOT been initialized");
		err = rp_mutex_unlock(&(cqueue->mutex));
		return err;			
	}
	if (!tqueue->ntask)	
	{
		rp_log(LOG_DEBUG, fname, "The task queue is empty");
		err = rp_mutex_unlock(&(tqueue->mutex));
		return err;			
	}
	
	tnode = tqueue->tail;
	while (tnode->prev)
	{
		tnode = tnode->prev;
		free(tnode->next);
	}
	free(tnode);

	tqueue->head = NULL;
	tqueue->tail = NULL;
	tqueue->qlength = 0;
	tqueue->ntask = 0;
	
	err = rp_mutex_unlock(&(tqueue->mutex));
	
	return err;
}


Err rp_tqueue_put(Tqueue *tqueue, Task *task)
{
	char fname[] = "rc_tqueue_put";
	char msg[] = "fatal error";
	Err err;
	Tnode *tnode;
     	
	err = rp_mutex_lock(&(tqueue->mutex));
	if (!tqueue->qlength)	
	{
		rp_log(LOG_DEBUG, fname, "The task queue has NOT been initialized");
		err = rp_mutex_unlock(&(cqueue->mutex));
		return err;			
	}
	if (tqueue->ntask == tqueue->qlength)	
	{
		rp_log(LOG_INFO, fname, "The task queue is full");
		err = rp_mutex_unlock(&(tqueue->mutex));
		return err;			
	}
	
	if (tqueue->ntask == 0)
	{
		tnode = (Tnode*)malloc(sizeof(Tnode));
		tnode->next = NULL;
		tnode->prev = NULL;
		memcpy(&(tnode->task), task, sizeof(Task));
		
		tqueue->head = tnode;
		tqueue->tail = tnode;
		tqueue->ncmd++;
	} else {		// 1<=ncmd<qlength
		tnode = (Tnode*)malloc(sizeof(Tnode));
		tnode->next = NULL;
		tnode->prev = tqueue->tail;
		memcpy(&(tnode->task), task, sizeof(Task));
				
		tqueue->tail->next = tnode;
		tqueue->tail = tnode;
		tqueue->ntask++;
		
	}
	
	err = rp_mutex_unlock(&(cqueue->mutex));
}

Task* rp_tqueue_get(Tqueue *tqueue)
{
	char fname[] = "rc_tqueue_get";
	char msg[] = "fatal error";
	Err err;	
	Tnode *tnode;
	Task *task;

	err = rp_mutex_lock(&(tqueue->mutex));
	if (!tqueue->qlength)	
	{
		rp_log(LOG_DEBUG, fname, "The task queue has NOT been initialized");
		err = rp_mutex_unlock(&(cqueue->mutex));
		return err;			
	}
	if (!tqueue->ntask)	
	{
		rp_log(LOG_INFO, fname, "The task queue is full");
		err = rp_mutex_unlock(&(tqueue->mutex));
		return err;			
	}
	
	tnode = tqueue->head;
	task = (Task*)malloc(sizeof(Task));
	memcpy(task, tnode->task);
	
	if (tqueue->tail == tqueue->head) 
	{
		tqueue->head = NULL;
		tqueue->tail = NULL;
	} else {
		tqueue->head = tqueue->head->next;
		tqueue->head->prev = NULL;
	}
	free(tnode);
	
	err = rp_mutex_unlock(&(tqueue->mutex));

	return task;
}


Err rp_ptable_init(Ptable *ptable)
{
	char fname[] = "rp_ptable_cleanup";
	char msg[] = "fatal error";
	Err err;

	err = rp_mutex_lock(&(ptable->mutex));
	ptable->count = 0;
	ptable->head = NULL;
	ptable->tail = NULL;
	err = rp_mutex_unlock(&(ptable->mutex));

	return err;
}

Err rp_ptable_finalize(Ptable *ptable)
{
	char fname[] = "rp_ptable_cleanup";
	char msg[] = "fatal error";
	Err err;

	err = rp_mutex_lock(&(ptable->mutex));

	// check pid and sigterm the process
	Pnode *pnode, *temp;
	int *status;

	pnode = ptable->head;
	while (pnode)
	{
		if (!waitpid(pnode->pid, status, WHOHANG))
		{	/*	the process is still alive	  */
			err = rp_process_terminate(pnode->pid, SIGTERM);
		} else {
			if (WIFEXITED(status))
			{
				WEXITSTATUS(status);				TODO: check rpointd exit status
				rp_log(LOG_DEBUG, fname, "Rpointd PID=%d exits with the status");
			} else if (WIFSIGNALED(status))
			{
				WTERMSIG(status);
				rp_log(LOG_DEBUG, fname, "Rpointd PID=%d exits with the signal");
			} else {
				rp_log(LOG_CRITICAL, fname, "Rpointd PID=%d exits with unknown cause");
			}
		}
		
		temp = pnode;
		pnode = pnode->next;
		/*	unmap & close the file   */
		if (!mumap((void*)&(temp->score), sizeof(Score)))
		{
			rp_log(LOG_CRITICAL, fname, "Cannot unmap the file under /tmp/rally");
		}  else {
			free(temp->score);
		}
		if (close(temp->fd) < 0)
		{
			rp_log(LOG_CRITICAL, fname, "Cannot close the file under /tmp/rally");
		}		
		free(temp);
	} 

	ptable->head = NULL;
	ptable->tail = NULL;
	ptable->count = 0;
	rp_log(LOG_DEBUG, fname, "Process Table has been finalized");

	err = rp_mutex_unlock(&(ptable->mutex));

	return err;
}

Err rp_ptable_put(Ptable *ptable, Pnode *pnode)
{
	char fname[] = "rp_ptable_put";
	char msg[] = "fatal error";
	Err err;

	err = rp_mutex_lock(&(ptable->mutex));

	if (!ptable->count++)
	{
		pnode->next = NULL;
		pnode->prev = NULL;
		ptable->head = pnode;
		ptable->tail = pnode;
	} else {
		pnode->next = NULL;
		pnode->prev = ptable->tail;
		ptable->tail->next = pnode;
		ptable->tail = pnode;
	}

	err = rp_mutex_unlock(&(ptable->mutex));
	return err;
}


Err rp_ptable_cleanup(Ptable *ptable)
{
	char fname[] = "rp_ptable_cleanup";
	char msg[] = "fatal error";
	Err err;

	err = rp_mutex_lock(&(ptable->mutex));
	if (!ptable->count) 
	{
		rp_log(LOG_DEBUG, fname, "No pending process to be shutdown");
		err = rp_mutex_unlock(&(ptable->mutex));
		return err;
	}

	// check pid and sigterm the process
	Pnode *pnode = ptable->head;
	int *status;
	
	while (!pnode)
	{
		if (!waitpid(pnode->pid, status, WHOHANG))
		{	/*	the process is still there	  */
			err = rp_process_terminate(pnode->pid, SIGTERM);
		} else {
			if (WIFEXITED(status))
			{
				WEXITSTATUS(status);				TODO: check rpointd exit status
				rp_log(LOG_DEBUG, fname, "Rpointd PID=%d exits with the status");
			} else if (WIFSIGNALED(status))
			{
				WTERMSIG(status);
				rp_log(LOG_DEBUG, fname, "Rpointd PID=%d exits with the signal");
			} else {
				rp_log(LOG_CRITICAL, fname, "Rpointd PID=%d exits with unknown cause");
			}
			
			/*	remove the pnode from ptable   */			
			if (ptable->count == 1)
			{
				ptable->head = NULL;
				ptable->tail = NULL;
			} else if (pnode == ptable->head)
			{
				ptable->head = ptable->head->next;
				ptable->head->prev = NULL;
			} else if (pnode == ptable->tail)
			{
				ptable->tail = ptable->tail->prev;
				ptable->tail->next = NULL;
			} else {
				pnode->prev->next = pnode->next;
				pnode->next->prev = pnode->prev;
			}
			ptable->count--;
			
			/*	unmap & close the file   */
			if (!mumap((void*)&(snode->score), sizeof(Score)))
			{
				rp_log(LOG_CRITICAL, fname, "Cannot unmap the file under /tmp/rally");
			}  else {
				free(temp->score);
			}
			if (close(snode->fd) < 0)
			{
				rp_log(LOG_CRITICAL, fname, "Cannot close the file under /tmp/rally");
			}
			free(pnode);	
		}
		pnode = pnode->next;
	} 

	err = rp_mutex_unlock(&(ptable->mutex));
	return err;
}

/*
	if (!mumap((void*)&(snode->score), sizeof(Score)))
	{
		rp_log(LOG_CRITICAL, fname, "Cannot unmap the file under /tmp/rally");
	} else {
		close(snode->fd);
	}

			
			if (!mumap((void*)&(snode->score), sizeof(Score)))
			{
				rp_log(LOG_CRITICAL, fname, "Cannot unmap the file under /tmp/rally");
			} else {
				close(snode->fd);
				snode->fd = 0;
			}
*/




